package com.hliushi.spark.rdd

import org.apache.commons.lang3.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/15 14:02
 */
class StagePractice {


  val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("stage_practice")

  val sc = new SparkContext(conf)

  /**
   * 需求: 以年月为基础, 进行reduceByKey统计东四地区的PM
   */
  @Test
  def pmProcess(): Unit = {
    // 1.读取文件
    // bug点  问题出现: 因为BeijingPM20100101_20151231.csv 文件第一行带有字段名称, 没有选择过滤第一行数据,从而出现错误
    //  解决方案1: 把数据源替换成没有表头模式[no_header]的
    val source = sc.textFile("dataset/BeijingPM20100101_20151231.csv")

    //  解决方案2:使用spark中的CSV的解析器 -- 不过需要导入使用到spark-sql的jar包
    //df = spark.read.format("csv").option("header", "true").option("mode", "DROPMALFORMED").csv("hdfs:///shuju/taobao.csv")

    // 解决方案3: 使用mapPartitionsWithIndex算子  --> 疑问:不知道首行是否永远在Partition 0？


    // 2.抽取需要的列

    // 通过算子来处理数据
    //    2.0.[可选] 如果选择带有表头的CSV数据, 则把第一行数据过滤掉, 如果选择no_header, 则不需要
    //    2.1.抽取数据,  返回元组类型:  ((年, 月), PM)
    //    2.2.清洗  过滤掉空的字符串, 过滤掉NA,  x._2 != "NA" 替换成 !x._2.equalsIgnoreCase("NA")
    //    2.3.聚合
    //    2.4.排序
    // ------------------------------------------
    val filterData = source
      .mapPartitionsWithIndex((ix: Int, it: Iterator[String]) => {
        if (ix == 0) it.drop(1)
        it
      })
      .map(x => {
        val arr = x.split(",")
        ((arr(1), arr(2)), arr(6))
      }).filter(x => StringUtils.isNotEmpty(x._2) && !x._2.equalsIgnoreCase("NA"))
      .map(x => x._1 -> x._2.toInt)
      .reduceByKey((total: Int, curr: Int) => total + curr)
      .sortBy(x => x._2, ascending = false)


    // 3.获取结果
    println(s"filterData.count() = ${filterData.count()}")
    filterData.take(10).foreach(println(_))

    // 4.关闭资源
    sc.stop()
  }


}